
#include <iostream>

#include <stdio.h>
#include <readline/readline.h>
#include <readline/history.h>

#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/tokenizer.hpp>

#include <xbase/xbase_utility.hpp>
#include <xbase/client_environment.hpp>
#include <xbase/sync_handler.hpp>
#include <xbase/io_service_pool_impl.hpp>
#include <xbase/compressor.hpp>

#include "../common/logger.hpp"

class shell_env
{
public:

	typedef shell_env this_type;
	
	typedef std::map<std::string, xbase::row_mutation> row_map;
	typedef boost::shared_ptr<row_map> row_map_ptr;
	typedef std::pair<xbase::protocol::table_info_ptr, row_map_ptr> table_item;
	typedef std::map<std::string, table_item> table_map;

	typedef std::vector<xbase::row_result> row_vector;
	typedef boost::shared_ptr<row_vector> row_vector_ptr;

public:
	~shell_env() {
		if(m_xbase_env_ptr)
		{
			wait_for_pending_ops();
			m_xbase_env_ptr.reset();
		}
		xbase::async_module_stats::print(std::cout);
	}

	void set_up(const char *master_ip,
		    short master_port,
		    bool enable_async_module_stats = true) {
		if(enable_async_module_stats)
			xbase::async_module_stats::enable();

		xbase::client_options_ptr options(new xbase::client_options);
		xbase::io_service_pool_ptr pool(new xbase::impl::single_io_service);
		const boost::asio::ip::tcp::endpoint master_addr(
			boost::asio::ip::address::from_string(master_ip),
			master_port
			);

		//m_xbase_env_ptr.reset(new xbase::client_environment(options, pool, master_addr));
		m_xbase_env_ptr.reset(new xbase::client_environment(options, pool, master_ip, master_port));

		boost::mutex::scoped_lock lock(m_pending_mutex);
		m_pending_ops = 0;
		m_complete_info.clear();
	}

	void wait_for_pending_ops() {
		boost::mutex::scoped_lock lock(m_pending_mutex);
		while(m_pending_ops != 0)
		{
			while(! m_complete_info.empty())
			{
				std::cout << m_complete_info.front();
				m_complete_info.pop_front();
				--m_pending_ops;
			}
			if(m_pending_ops == 0)
				break;

			lock.unlock();
			
			std::cout << "waiting for pending ops " << m_pending_ops
				  << " ..." << std::endl;
			xbase::async_module_stats::print(std::cout);
			sleep(3);

			lock.lock();
		}
	}

	void enable_async_module_stats() {
		xbase::async_module_stats::enable();
		xbase::async_module_stats::clear();
	}

	void disable_async_module_stats() {
		xbase::async_module_stats::clear();
		xbase::async_module_stats::disable();
	}

	void print_complete_info() {
		boost::mutex::scoped_lock lock(m_pending_mutex);
		std::size_t count = 0;
		if(! m_complete_info.empty())
			std::cout << "--------------------------" << std::endl;
		while(! m_complete_info.empty())
		{
			std::cout << m_complete_info.front();
			m_complete_info.pop_front();
			--m_pending_ops;
			++count;
		}
		if(count != 0)
		{
			std::cout << "--- " << count
				  << " ops completed ---"
				  << std::endl;
		}
	}

	void list_table() {
		add_pending_ops();
		xbase::table_lister lister;
		m_xbase_env_ptr->async_apply(lister,
					     boost::bind(&this_type::on_table_listed,
							 this, _1, _2));
	}

	void drop_table(const std::string &table_name) {
		add_pending_ops();
		xbase::table_deletion deletion(table_name);
		m_xbase_env_ptr->async_apply(deletion,
					     boost::bind(&this_type::on_table_dropped,
							 this, table_name,
							 _1));
	}
	
	void create_table(const xbase::table_creation &creation) {
		add_pending_ops();
		m_xbase_env_ptr->async_apply(creation,
					     boost::bind(&this_type::on_table_created,
							 this, creation.table_info().m_table_name,
							 _1));
	}

	void add_row_mutation(const std::string &table_name,
			      const std::string &row,
			      const std::string &column,
			      const std::string &value,
			      uint64_t timestamp) {
		table_map::iterator t_it = m_tables.find(table_name);
		if(t_it == m_tables.end())
		{
			// 异步接口同步化
			xbase::protocol::table_info_ptr table_info;
			xbase::table_info_cache::sync_handler_type sh;
			m_xbase_env_ptr->get_table_info_cache().async_get_table_info(
				table_name,
				sh.make_handler()
				);
			sh.wait();
			if(sh.get_error_code())
			{
				std::cout << "error: " << sh.get_error_code() << " - " << sh.get_error_code().message() << std::endl;
				return;
			}
			table_info = sh.get_table_info();

			table_item item(table_info, row_map_ptr(new row_map()));
			std::pair<table_map::iterator, bool> ret =
				m_tables.insert(table_map::value_type(table_name, item));
			assert(ret.second == true);
			t_it = ret.first;
		}

		row_map &rows = *(t_it->second.second);
		row_map::iterator it = rows.find(row);
		if(it == rows.end())
		{
			std::pair<row_map::iterator, bool> ret =
				rows.insert(row_map::value_type(row, xbase::row_mutation(t_it->second.first, row)));
			assert(ret.second == true);
			it = ret.first;
		}
		xbase::row_mutation &mutation = it->second;
		boost::system::error_code ec =
			mutation.add(column, value, timestamp);
		if(ec)
		{
			std::cout << "add row \"" << row
				  << "\" failed: " << ec.message() << std::endl;
		}
	}

	void apply(const std::string &table_name,
		   const std::string &row,
		   uint64_t timestamp) {
		std::deque<table_map::value_type> row_queue;

		if(table_name.empty()) // apply all table
		{
			for(table_map::iterator it = m_tables.begin();
			    it != m_tables.end(); ++it)
			{
				row_queue.push_back(*it);
			}
			m_tables.clear();
		}
		else
		{
			const table_map::iterator it = m_tables.find(table_name);
			if(it != m_tables.end())
			{
				row_queue.push_back(*it);
				m_tables.erase(it);
			}
		}

		std::deque<xbase::row_mutation> ptr_q;

		while(! row_queue.empty())
		{
			table_map::value_type item = row_queue.front();
			row_queue.pop_front();
			
			row_map &rows = *item.second.second;
			
			if(row.empty()) // apply all row
			{
				for(row_map::iterator it = rows.begin();
				    it != rows.end(); ++it)
				{
					ptr_q.push_back(it->second);
				}
				rows.clear();
			}
			else
			{
				const row_map::iterator it = rows.find(row);
				if(it != rows.end())
				{
					ptr_q.push_back(it->second);
					rows.erase(it);
				}
			}

			if(! item.second.second->empty()) // 如果数据还有剩余，则再放回到table_map中
			{
				std::pair<table_map::iterator, bool> ret =
					m_tables.insert(item);
				assert(ret.second == true);
			}
		}

		if(ptr_q.empty())
		{
			std::cout << "not data"
				  << (row.empty() ? "" : " for row \"")
				  << row << '"'
				  << (table_name.empty() ? "" : " in table \"")
				  << table_name << '"'
				  << std::endl;
		}

		while(! ptr_q.empty())
		{
			xbase::row_mutation mutation = ptr_q.front();
			ptr_q.pop_front();
			
			if(mutation.empty())
			{
				std::cout << "no mutations for row \""
					  << mutation.row_name()
					  << "\" in table \""
					  << mutation.table_name() << '"'
					  << std::endl;
				continue;
			}
			
			add_pending_ops();
			
			m_xbase_env_ptr->async_apply(mutation,
						     timestamp,
						     boost::bind(&this_type::on_apply, this,
								 mutation,
								 _1));
		}
	}

	void scan(const xbase::scan_info &option) {
		add_pending_ops();
		row_vector_ptr vec_ptr(new row_vector);
		xbase::scanner_ptr ptr = m_xbase_env_ptr->new_scanner(option);
		ptr->async_get_next_row(boost::bind(&this_type::on_got_next_row,
						    this, ptr, vec_ptr, option,
						    _1, _2));
	}

protected:
	void on_table_listed(std::vector<std::string> vec,
			     const boost::system::error_code &ec) {
		std::ostringstream out;
		if(ec)
		{
			out << "list tables failed: " << ec.message() << std::endl;
		}
		else
		{
			out << "table list:" << std::endl;
			for(std::size_t i=0; i<vec.size(); ++i)
			{
				out << '\t' << vec[i] << std::endl;
			}
			out << "total " << vec.size() << " tables" << std::endl;
		}
		complete_pending_ops(out.str());
	}

	void on_got_next_row(xbase::scanner_ptr ptr, row_vector_ptr rows,
			     xbase::scan_info option,
			     xbase::row_result_ptr row,
			     const boost::system::error_code &ec) {
		if(ec)
		{
			std::ostringstream out;
			std::size_t count_row = rows->size();
			std::size_t count_record = 0;
			if(! rows->empty())
			{
				for(std::size_t i=0; i<rows->size(); ++i)
				{
					const xbase::row_result &vec = (*rows)[i];
					out << dump_string_char(vec.m_row_name) << std::endl;
					for(std::size_t j=0; j<vec.m_cell_vector.size();++j)
					{
						const xbase::cell &c = vec.m_cell_vector[j];
						out << "\t" << c.m_column_name
						    << " versions(" << c.get_version_count() << ")" << std::endl;
						for(std::size_t k=0; k<c.get_version_count(); ++k)
						{
							xbase::cell::version v;
							c.get_version(k, v);
							out << "\t\t" << v.first
							    << " (" << dump_timestamp(v.first) << ")\t"
							    << (v.second.empty()
								? std::string("#<deletion record>")
								: dump_string_char(v.second))
							    << std::endl;
							++count_record;
						}
					}
				}
			}

			out << "---------" << std::endl
			    << "got " << count_row << " rows with "
			    << count_record << " records in range [\""
			    << option.m_start_row
			    << "\", \"" << option.m_end_row
			    << "\")" << std::endl;

			if(ec != xbase::error::eos)
			{
				out << "scan failed: " << ec.message() << std::endl;
			}

			complete_pending_ops(out.str());
		}
		else
		{
			rows->push_back(*row);
			ptr->async_get_next_row(boost::bind(&this_type::on_got_next_row, this,
							    ptr, rows, option, _1, _2));
		}
	}

	void on_apply(xbase::row_mutation mutation,
		      const boost::system::error_code &ec) {
		std::ostringstream out;
		if(ec)
		{
			out << "writting row \"" << mutation.row_name()
			    << "\" into table \"" << mutation.table_name()
			    << "\" failed: " << ec.message() << std::endl;
		}
		else
		{
			out << "row \"" << mutation.row_name()
			    << "\" written into table \"" << mutation.table_name()
			    << "\"" << std::endl;
		}
		complete_pending_ops(out.str());
	}

	void on_table_created(const std::string &table_name,
			      const boost::system::error_code &ec) {
		std::ostringstream out;
		if(ec)		// error occur
		{
			out << "creating table \"" << table_name
			    << "\" failed: "
			    << ec.message() << std::endl;
		}
		else
		{
			out << "table \"" << table_name
			    << "\" created" << std::endl;
		}
		complete_pending_ops(out.str());
	}

	void on_table_dropped(const std::string &table_name,
			      const boost::system::error_code &ec) {
		std::ostringstream out;
		if(ec)		// error occur
		{
			out << "dropping table \"" << table_name
			    << "\" failed: "
			    << ec.message() << std::endl;
		}
		else
		{
			out << "table \"" << table_name
			    << "\" dropped" << std::endl;
		}
		complete_pending_ops(out.str());
	}

	void add_pending_ops() {
		boost::mutex::scoped_lock lock(m_pending_mutex);
		++m_pending_ops;
	}

	void complete_pending_ops(const std::string &info) {
		boost::mutex::scoped_lock lock(m_pending_mutex);
		m_complete_info.push_back(info);
	}
private:
	xbase::client_environment_ptr m_xbase_env_ptr;

	table_map m_tables;

	boost::mutex m_pending_mutex;
	std::size_t m_pending_ops;
	std::deque<std::string> m_complete_info;
};

class shell_vm
{
public:
	typedef boost::escaped_list_separator<char> sep;
	typedef boost::tokenizer<sep> tok;

public:
	shell_vm(shell_env &env)
		: m_env(env)
		, m_input_sep("\\", // ?
			      ", ",  // ?
			      "\"" // ?
			) {}

	void run();

protected:
	void command_error() {
		std::cout << "error: bad command!" << std::endl
			  << "command list:" << std::endl
			  << "\tdrop <table name>" << std::endl
			  << "\tcreate <table_name> <schema_info>" << std::endl
			  << "\t\tschema_info: <schema_item> [schema_item]*" << std::endl
			  << "\t\tschema_item: <family_item|group_item>" << std::endl
			  << "\t\tgroup_item: <group_name> <compress_method> <cache_rate> <block_size> <max_num_sstable> <max_size_sstable> <max_size_memtable>" << std::endl
			  << "\t\tfamily_item: <family_name> <group_name> <max_version_number> <max_ttl> <deadtime>" << std::endl
			  << "\t\tcompress_method: <none|quicklz|zlib|minilzo>" << std::endl
			  << "\tadd <table name> <row> <column> [value] [timestamp]" << std::endl
			  << "\tapply [table name] [row] [timestamp]" << std::endl
			  << "\tscan <table name> <start row> <end row> <column1[;columns2;...]> [latest timestamp] [oldest_timestamp] [ttl] [version_number] [option] [num row]" << std::endl
			  << "\tlist" << std::endl
			  << "\tsync" << std::endl
			  << "\tquit" << std::endl;
	}

private:
	shell_env &m_env;
	
	sep m_input_sep;
};

int main(int argc, char **argv)
{
	if(argc != 3)
	{
		std::cout << "test_row_mutator <master ip> <master port>" << std::endl;
		return 0;
	}

	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	shell_env env;
	env.enable_async_module_stats();
	env.set_up(argv[1], ::atoi(argv[2]));

	shell_vm vm(env);
	vm.run();

	std::cout << std::endl;

	return 0;
}

void save_history_list(const std::string &history_file,
		       std::size_t max_size,
		       std::vector<std::string> &history_list)
{
	std::ofstream out(history_file.c_str(), std::ios::trunc);
	if(out)
	{
		typedef std::vector<std::string>::iterator it_t;
		it_t end = history_list.end();
		it_t start = history_list.begin();
		if(history_list.size() > max_size)
		{
			start = end - max_size;
		}
		for(it_t it = start; it != end; ++it)
		{
			out << *it << std::endl;
		}
		out.close();
		if(history_list.size() > (max_size * 2))
		{
			history_list.erase(history_list.begin(),
					   start);
		}
	}
	else
	{
		std::cout << "open history file " << history_file
			  << "failed, so discard history list" << std::endl;
	}
}

template<typename T>
inline
bool parse_integer(T &ts, const std::string &str)
{
	try {
		ts = boost::lexical_cast<T>(str);
		return true;
	} catch (const std::exception &e) {
		return false;
	}
}

inline
bool parse_columns(std::vector<std::string>&columns, std::string str)
{
	if(str.empty())
	{
		return false;
	}
	
	typedef boost::tokenizer<boost::escaped_list_separator<char> > tok;
	boost::escaped_list_separator<char> separator("\\", ";", "\";");
	tok t(str, separator);
	for(tok::iterator it = t.begin(); it != t.end(); ++it)
	{
		columns.push_back(*it);
	}
	if(columns.empty())
	{
		return false;
	}
	return true;
}

void shell_vm::run()
{
	const char *home = ::getenv("HOME");
	if(home == NULL)
		home = "/tmp";
	std::string history_file(home);
	history_file += "/.xbaseshell_history";

	std::vector<std::string> history_list;
	const std::size_t history_size = 10000;
	bool history_enabled = true;

	int fd = -1;
	if(::access(history_file.c_str(), F_OK) != 0)
	{
		fd = ::creat(history_file.c_str(),
			     S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
		if(fd < 0)
		{
			std::cout << "create history file " << history_file
				  << " failed, disable history logging" << std::endl;
			history_enabled = false;
		}
		::close(fd);
	}
	else
	{
		FILE *fp = ::fopen(history_file.c_str(), "r");
		if(fp == NULL)
		{
			std::cout << "open history file " << history_file
				  <<" failed, disable history logging" << std::endl;
			history_enabled = false;
		}
		else		// load history
		{
			const std::size_t max_line_size = 64 * 1024;
			char line[max_line_size + 1];
			while(::fgets(line, max_line_size, fp))
			{
				if(line[strlen(line) - 1] == '\n')
					line[strlen(line) - 1] = '\0';
				add_history(line);
				history_list.push_back(line);
			}
			::fclose(fp);
		}
	}

	for(char *input_str = readline("> "); input_str != NULL; free(input_str), input_str = readline("> "))
	{
		m_env.print_complete_info();

		if((*input_str == '\0') || // 空行
		   (*input_str == '#'))    // 注释行
		{
			continue;
		}

		const std::string input(input_str);
		tok t(input, m_input_sep);

		// for(tok::iterator it = t.begin();
		//     it != t.end(); ++it)
		// 	std::cout << "\"" << *it << "\",";
		// std::cout << std::endl;

		tok::iterator it = t.begin();
		if(it == t.end()) // empty
		{
			continue;
		}
		else if(*it == "quit") // quit
		{
			break;
		}
		else		// ok, and save into history list
		{
			add_history(input_str);
			history_list.push_back(input);
			save_history_list(history_file,
					  history_size,
					  history_list);
		}

		if(*it == "drop")
		{
			std::string table_name;
			++it;
			if(it != t.end())
			{
				table_name = *it;
			}

			if(table_name.empty())
			{
				command_error();
				continue;
			}

			m_env.drop_table(table_name);
		}
		else if(*it == "create")
		{
			std::string table_name;

			++it;
			if(it != t.end())
			{
				table_name = *it;
			}

			if(table_name.empty())
			{
				command_error();
				continue;
			}
			++it;

			xbase::table_creation creation(table_name);

			bool error = false;
			for(;it != t.end(); ++it)
			{
				if(*it == "group")
				{
					std::string name;
					xbase::table_creation::group_options op;
					++it;
					if(it != t.end())
					{
						name = *it;
						++it;
					}
					if(it != t.end())
					{
						if(*it == "none")
							op.m_compress_method = xbase::compressor::NONE;
						else if(*it == "quicklz")
							op.m_compress_method = xbase::compressor::QUICKLZ;
						else if(*it == "zlib")
							op.m_compress_method = xbase::compressor::ZLIB;
						else if(*it == "minilzo")
							op.m_compress_method = xbase::compressor::MINILZO;
						else
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_cache_rate, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_block_size, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_max_num_sstable, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_max_size_sstable, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_max_size_memtable, *it))
						{
							error = true;
							break;
						}
					}
					if(it == t.end())
					{
						error = true;
						break;
					}
					boost::system::error_code ec = creation.add_group(name, op);
					if(ec)
					{
						std::cout << ec.message() << std::endl;
						error = true;
						break;
					}
				}
				else if(*it == "family")
				{
					std::string name;
					std::string gname;
					xbase::table_creation::family_options op;

					// avoid warnning
					op.m_max_num_version = 0;
					op.m_max_ttl = 0;
					op.m_dead_time = 0;
					op.version_option = 0;
					op.family_options = 0;

					++it;
					if(it != t.end())
					{
						name = *it;
						++it;
					}
					if(it != t.end())
					{
						gname = *it;
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_max_num_version, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_max_ttl, *it))
						{
							error = true;
							break;
						}
						++it;
					}
					if(it != t.end())
					{
						if(! parse_integer(op.m_dead_time, *it))
						{
							error = true;
							break;
						}
					}
					if(it == t.end())
					{
						error = true;
						break;
					}
					boost::system::error_code ec = creation.add_family(name, gname, op);
					if(ec)
					{
						std::cout << ec.message() << std::endl;
						error = true;
						break;
					}
				}
				else
				{
					error = true;
					break;
				}
			}
			if(error)
			{
				command_error();
				continue;
			}
			m_env.create_table(creation);
		}
		else if(*it == "add")
		{
			std::string table_name;
			std::string row;
			std::string column;
			std::string value;
			uint64_t timestamp = 0;
			

			++it;
			if(it != t.end())
			{
				table_name = *it;
				++it;
			}
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			if(it != t.end())
			{
				column = *it;
			}
			// 上面是必写的
			if(it == t.end())
			{
				command_error();
				continue;
			}
			// 这是可选的
			++it;
			if(it != t.end())
			{
				value = *it;
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(timestamp, *it))
				{
					timestamp = 0;
				}
				++it;
			}

			if(table_name.empty())
			{
				std::cout << "error: " << xbase::error::bad_table_name << std::endl;
				continue;
			}
			if(row.empty())
			{
				std::cout << "error: " << xbase::error::bad_row_name << std::endl;
				continue;
			}

			m_env.add_row_mutation(table_name, row, column, value, timestamp);
		}
		else if(*it == "apply")
		{
			std::string table_name;
			std::string row;
			uint64_t timestamp = cur_micro_seconds();

			++it;
			if(it != t.end())
			{
				table_name = *it;
				++it;
			}
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(timestamp, *it))
				{
					timestamp = cur_micro_seconds();
				}
			}

			m_env.apply(table_name, row, timestamp);
		}
		else if(*it == "scan")
		{
			xbase::scan_info option;

			++it;
			if(it != t.end())
			{
				option.m_table_name = *it;
				++it;
			}
			if(it != t.end())
			{
				option.m_start_row = *it;
				++it;
			}
			if(it != t.end())
			{
				option.m_end_row = *it;
				++it;
			}
			if(it != t.end())
			{
				if(!parse_columns(option.m_column_names,
						  *it))
				{
					command_error();
					continue;
				}
			}
			// 上面是必写的
			if(it == t.end())
			{
				command_error();
				continue;
			}
			// 这是可选的
			++it;
			if(it != t.end())
			{
				if(! parse_integer(option.m_latest_time_stamp,
						   *it))
				{
					command_error();
					continue;
				}
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(option.m_oldest_time_stamp,
						   *it))
				{
					command_error();
					continue;
				}
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(option.m_ttl, *it))
				{
					command_error();
					continue;
				}
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(option.m_version_number,
					       *it))
				{
					command_error();
					continue;
				}
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(option.m_options, *it))
				{
					command_error();
					continue;
				}
				++it;
			}
			if(it != t.end())
			{
				if(! parse_integer(option.m_num_row, *it))
				{
					command_error();
					continue;
				}
				++it;
			}

			m_env.scan(option);
		}
		else if(*it == "list")
		{
			m_env.list_table();
		}
		else if(*it == "sync")
		{
			m_env.wait_for_pending_ops();
		}
		else
		{
			command_error();
		}
	}
}

